package com.gitee.xhs.saveAndModifyingPointWriter.keyedState.mapState;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.dao.ModifyingPointDao;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import com.gitee.xhs.util.SavePointApiUtil;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.state.api.OperatorTransformation;
import org.apache.flink.state.api.SavepointWriter;
import org.apache.flink.state.api.StateBootstrapTransformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static com.gitee.xhs.StateConstant.CHECK_POINT_BASE_PATH;
import static com.gitee.xhs.keyedState.mapState.MapStateConstant.CUSTOM_MAP_STATE_DIR;
import static com.gitee.xhs.keyedState.mapState.MapStateConstant.MAP_STATE_UID;
import static com.gitee.xhs.saveAndModifyingPointWriter.SavePointWriterConstant.MAX_PARALLELISM;
import static com.gitee.xhs.saveAndModifyingPointWriter.SavePointWriterConstant.MODIFYING_UID_PREFIX;

/**
 * @author BigHead
 * @date 2022/7/12 11:08
 * @description MapStateNewSavePointsMain
 */
public class MapStateNewSavePointsModifyingMain {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.simpleEnv(new EnvironmentConfig(null));

        DataStream<Transaction> transactionDataStream = streamExecutionEnvironment.fromElements(
                new Transaction(1, 0, 0.1),
                new Transaction(2, 0, 0.1),
                new Transaction(3, 0, 0.1),
                new Transaction(4, 0, 0.1),
                new Transaction(5, 0, 0.1)
        );
        String newSavePointDirPathName = SavePointApiUtil.getNewSavePointWriterJobPath(CHECK_POINT_BASE_PATH + CUSTOM_MAP_STATE_DIR);
        StateBootstrapTransformation<Transaction> transformation = OperatorTransformation
                .bootstrapWith(transactionDataStream)
                .keyBy(Transaction::getAccountId)
                .transform(new MapStateNewSavePointsFunction());
        SavepointWriter
                .newSavepoint(new HashMapStateBackend(), MAX_PARALLELISM)
                .withOperator(MAP_STATE_UID, transformation)
                .write(newSavePointDirPathName);

        ModifyingPointDao modifyingPointDao = SavePointApiUtil.getModifyingPath(CHECK_POINT_BASE_PATH + CUSTOM_MAP_STATE_DIR);
        SavepointWriter
                .fromExistingSavepoint(modifyingPointDao.getOldPath(), new HashMapStateBackend())
                .withOperator(MAP_STATE_UID + MODIFYING_UID_PREFIX, transformation)
                .write(modifyingPointDao.getNewPath());

        streamExecutionEnvironment.execute("MapStateNewSavePointsMain");
    }

}
